package com.etrans.web.gps.business.command.controller;

import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.etrans.egov.base.cmd.obj.CmdArgsPostQueryReq;
import com.etrans.egov.base.cmd.obj.CmdIds;
import com.etrans.egov.base.cmd.obj.ObjCmdReceipt;
import com.etrans.egov.base.cmd.obj.ObjCmdReply;
import com.etrans.egov.base.cmd.obj.ObjCmdRequest;
import com.etrans.lib.net2.base.tcp.aio.AioTcpClient;
import com.etrans.lib.net2.tcp.Net2AioClient;
import com.etrans.lib.net2.tcp.packet.Packet;
import com.etrans.lib.net2.tcp.packet.PacketPublish;
import com.etrans.lib.net2.tcp.packet.PacketRequest;
import com.etrans.lib.net2.tcp.packet.PacketResponse;
import com.etrans.lib.net2.tcp.packet.PacketSubscribeReply;
import com.etrans.lib.threading.ThreadUtils;
import com.etrans.lib.threading.ThreadingEvent;

public class Net2CmdProducer
{
	
	private static final Logger logger = LoggerFactory.getLogger(Net2CmdProducer.class);
	
	//-------------

	private Net2AioClient client;
	
	private ThreadingEvent connetedEvent = new ThreadingEvent(false);
	
	private ConcurrentHashMap<String, Callback> receiptCallbackMap = new ConcurrentHashMap<>();

	private ConcurrentHashMap<String, Callback> replyCallbackMap = new ConcurrentHashMap<>();
	
	public Net2CmdProducer(String _ip, int _port)
	{
		client = Net2AioClient.newNet2AioClient(_ip, _port, true);
		client.addListener(listener);
	}
	
	
	Net2AioClient.Listener listener = new Net2AioClient.Listener()
	{
		@Override
		public void onConnected(Net2AioClient client, AioTcpClient<Object> _sender) throws Exception
		{
			logger.info("onConnected: {}", client.getClient().getSessionKey());
			connetedEvent.set();
		}
		
		@Override
		public void onDisconnected(Net2AioClient client, AioTcpClient<Object> _sender, String _reason, Throwable _e) throws Exception
		{
			logger.info("onDisconnected: {}", client.getClient().getSessionKey());
			connetedEvent.clear();
		}

		@Override
		public void onReceivedResponse(Net2AioClient client, AioTcpClient<Object> _sender, PacketResponse response) throws Exception
		{
			logger.info("onReceivedResponse: {}", client.getClient().getSessionKey());
			ObjCmdReceipt objCmdReceipt = ObjCmdReceipt.makeFromJSONObject(JSON.parseObject(new String(response.getContent(), "utf-8")));
			handleObjCmdReceipt(objCmdReceipt);
		}

		@Override
		public void onReceivedRequest(Net2AioClient client, AioTcpClient<Object> _sender, PacketRequest request) throws Exception
		{
			logger.info("onReceivedRequest: {}", client.getClient().getSessionKey());
			ObjCmdReply objCmdReply = ObjCmdReply.makeFromJSONObject(JSON.parseObject(new String(request.getContent(), "utf-8")));
			handleObjCmdReply(objCmdReply);
		}
		
		@Override
		public void onReceivedSubscribeReply(Net2AioClient client, AioTcpClient<Object> _sender, PacketSubscribeReply reply) throws Exception
		{
			logger.info("onReceivedSubscribeReply: {}", client.getClient().getSessionKey());
		}
		
		@Override
		public void onReceivedPublish(Net2AioClient client, AioTcpClient<Object> _sender, PacketPublish publish) throws Exception
		{
			logger.info("onReceivedPublish: {}", client.getClient().getSessionKey());
		}
		
		@Override
		public void onReceivedPacket(Net2AioClient client, AioTcpClient<Object> _sender, Packet packet) throws Exception
		{
			logger.info("onReceivedPacket: {}", client.getClient().getSessionKey());
		}
	};

	public void push(ObjCmdRequest objCmdRequest, Callback callback) throws Exception
	{
		connetedEvent.waitIfFalse();
		
		if( callback!=null )
		{
			receiptCallbackMap.put(objCmdRequest.getUuid(), callback);
		}
		
		PacketRequest request = new PacketRequest();
		request.setContent(objCmdRequest.toJSONObject().toString().getBytes("utf-8"));
		client.sendRequest(request);
	}
	
	private void handleObjCmdReceipt(ObjCmdReceipt objCmdReceipt) throws Exception
	{
		Callback callback = receiptCallbackMap.remove(objCmdReceipt.getUuid());
		if( callback!=null )
		{
			callback.onObjCmdReceipt(objCmdReceipt);
			if( StringUtils.isNotBlank(objCmdReceipt.getReplyKey()) )
			{
				replyCallbackMap.put(objCmdReceipt.getReplyKey(), callback);
			}
		}
	}
	

	private void handleObjCmdReply(ObjCmdReply objCmdReply) throws Exception
	{
		Callback callback = replyCallbackMap.remove(objCmdReply.getReplyKey());
		if( callback!=null )
		{
			callback.onObjCmdReply(objCmdReply);
		}
	}

	public void start()
	{
		client.start();
		new Thread(checkTimeoutRunnable).start();
	}

	private Runnable checkTimeoutRunnable = new Runnable()
	{
		@Override
		public void run()
		{
			while(true)
			{
				ThreadUtils.sleep(1000);
				checkReceiptTimeout();
				checkReplyTimeout();
			}
		}
		
		private void checkReceiptTimeout()
		{
			for( Entry<String, Callback> entry : receiptCallbackMap.entrySet() )
			{
				String uuid = entry.getKey();
				Callback callback = entry.getValue();
				if( System.currentTimeMillis() > callback.receiptExpire )
				{
					try
					{
						callback = receiptCallbackMap.remove(uuid);
						if( callback!=null )
						{
							callback.onReceiptTimeout();
						}
					}
					catch (Exception e)
					{
						logger.error("", e);
					}
				}
			}
		}
		

		private void checkReplyTimeout()
		{
			for( Entry<String, Callback> entry : replyCallbackMap.entrySet() )
			{
				String uuid = entry.getKey();
				Callback callback = entry.getValue();
				if( System.currentTimeMillis() > callback.replyExpire )
				{
					try
					{
						callback = replyCallbackMap.remove(uuid);
						if( callback!=null )
						{
							callback.onReplyTimeout();
						}
					}
					catch (Exception e)
					{
						logger.error("", e);
					}
				}
			}
		}
	};
	
	/**
	 * 
	 */
	public static abstract class Callback
	{
		private long genTime;
		private long receiptExpire;
		private long replyExpire;
		
		public Callback(long receiptTimeout, long replyTimeout)
		{
			genTime = System.currentTimeMillis();
			receiptExpire = genTime + receiptTimeout;
			replyExpire   = genTime + replyTimeout;
		}
		
		public Callback()
		{
			this(5000, 10000);
		}
		
		public abstract void onObjCmdReceipt(ObjCmdReceipt objCmdReceipt) throws Exception;
		
		public abstract void onReceiptTimeout() throws Exception;

		public abstract void onObjCmdReply(ObjCmdReply objCmdReply) throws Exception;
		
		public abstract void onReplyTimeout() throws Exception;
		
		public long getGenTime()
		{
			return genTime;
		}
	}
	
	//--------
	
	public static void main(String[] args) throws Exception
	{
		Net2CmdProducer cmdProducer = new Net2CmdProducer("localhost", 7001);
		cmdProducer.start();
		
		CmdArgsPostQueryReq cmdArgsPostQueryReq = new CmdArgsPostQueryReq();
		cmdArgsPostQueryReq.setGnssCenterId(1);
		cmdArgsPostQueryReq.setObjectType((byte)11);
		cmdArgsPostQueryReq.setObjectId("22");
		cmdArgsPostQueryReq.setInfoId(33);
		cmdArgsPostQueryReq.setInfoContent("44");
		
		ObjCmdRequest objCmd = new ObjCmdRequest();
		objCmd.setCmdId(CmdIds.CMD_ID_POST_QUERY_REQ);
//		objCmd.setArgs(cmdArgsPostQueryReq.toJSONString().toString());
		
		cmdProducer.push(objCmd, new Net2CmdProducer.Callback()
		{
			@Override
			public void onObjCmdReceipt(ObjCmdReceipt objCmdReceipt) throws Exception
			{
				logger.info("onObjCmdReceipt: {}", objCmdReceipt.toJSONObject());
			}

			@Override
			public void onReceiptTimeout() throws Exception
			{
				logger.info("onReceiptTimeout");
			}

			@Override
			public void onObjCmdReply(ObjCmdReply objCmdReply) throws Exception
			{
				logger.info("onObjCmdReply: {}", objCmdReply.toJSONObject());
			}

			@Override
			public void onReplyTimeout() throws Exception
			{
				logger.info("onReplyTimeout");
			}
		});
	}

	public Net2AioClient getClient() {
		return client;
	}
	
}



